package com.atliuzu.app.dws;

import com.alibaba.fastjson.JSONObject;
import com.atliuzu.app.func.DimAsyncFunction;
import com.atliuzu.bean.CartAddUuBean;
import com.atliuzu.bean.TradeCourseBean;
import com.atliuzu.utils.ClickHouseUtil;
import com.atliuzu.utils.DateFormatUtil;
import com.atliuzu.utils.MyKafkaUtil;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.AsyncDataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.async.AsyncFunction;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;


import java.time.Duration;
import java.util.concurrent.TimeUnit;

/**
 *交易域类别-科目-课程-用户粒度下单各窗口汇总表
 * @Author zhang
 * @Time 2022-08-21-11:10
 */
public class DwsTradeCourseOrderWindow {
    public static void main(String[] args) throws Exception {
        //TODO 1.获取执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        // 1.1 状态后端设置
//        env.enableCheckpointing(3000L, CheckpointingMode.EXACTLY_ONCE);
//        env.getCheckpointConfig().setCheckpointTimeout(60 * 1000L);
//        env.getCheckpointConfig().setMinPauseBetweenCheckpoints(3000L);
//        env.getCheckpointConfig().enableExternalizedCheckpoints(
//                CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION
//        );
//        env.setRestartStrategy(RestartStrategies.failureRateRestart(
//                3, Time.days(1), Time.minutes(1)
//        ));
//        env.setStateBackend(new HashMapStateBackend());
//        env.getCheckpointConfig().setCheckpointStorage(
//                "hdfs://hadoop102:8020/ck"
//        );
//        System.setProperty("HADOOP_USER_NAME", "atguigu");

        //TODO 2.从kafka dwd_trade_order_detail主题读取数据
        DataStreamSource<String> streamSource = env.addSource(MyKafkaUtil.getFlinkKafkaConsumer(
                "dwd_trade_order_detail", "group1"));

        //TODO 3.将数据流转化为json
        SingleOutputStreamOperator<JSONObject> jsonObjDS =
                streamSource.flatMap(new FlatMapFunction<String, JSONObject>() {
            @Override
            public void flatMap(String value, Collector<JSONObject> out) throws Exception {
                JSONObject jsonObject = JSONObject.parseObject(value);
                out.collect(jsonObject);
            }
        });



        //TODO 4.将数据转化为javabean
        SingleOutputStreamOperator<TradeCourseBean> tradeCourseDS =
                jsonObjDS.flatMap(new FlatMapFunction<JSONObject, TradeCourseBean>() {
            @Override
            public void flatMap(JSONObject value, Collector<TradeCourseBean> out) throws Exception {
                TradeCourseBean build = TradeCourseBean.builder()
                        .courseId(value.getString("course_id"))
                        .courseName(value.getString("course_name"))
                        .userId(value.getString("user_id"))
                        .orderCount(1L)
                        .finalAmount(value.getDouble("final_amount"))
                        .couponReduce(value.getDouble("couponReduce"))
                        .originAmount(value.getDouble("origin_amount"))
                        .ts(DateFormatUtil.toTs(value.getString("create_time"), true)).build();

                out.collect(build);
            }
        });




        //TODO 5.设置水位线
        SingleOutputStreamOperator<TradeCourseBean> watermarksDS =
                tradeCourseDS.assignTimestampsAndWatermarks(WatermarkStrategy.<TradeCourseBean>forBoundedOutOfOrderness(Duration.ofSeconds(2)).withTimestampAssigner(new SerializableTimestampAssigner<TradeCourseBean>() {
                    @Override
                    public long extractTimestamp(TradeCourseBean element, long recordTimestamp) {
                        return element.getTs();
                    }
                }));


        //TODO 6.分组、开窗、聚合
        SingleOutputStreamOperator<TradeCourseBean> reduceDS = watermarksDS.keyBy(new KeySelector<TradeCourseBean,
                Tuple2<String, String>>() {
            @Override
            public Tuple2<String, String> getKey(TradeCourseBean value) throws Exception {
                return new Tuple2<>(value.getUserId(), value.getCourseId());
            }
        }).window(TumblingEventTimeWindows.of(Time.seconds(10))).reduce(new ReduceFunction<TradeCourseBean>() {
            @Override
            public TradeCourseBean reduce(TradeCourseBean value1, TradeCourseBean value2) throws Exception {
                value1.setOrderCount(value1.getOrderCount() + value2.getOrderCount());
                return value1;
            }
        }, new WindowFunction<TradeCourseBean, TradeCourseBean, Tuple2<String, String>, TimeWindow>() {
            @Override
            public void apply(Tuple2<String, String> stringStringTuple2, TimeWindow window,
                              Iterable<TradeCourseBean> input, Collector<TradeCourseBean> out) throws Exception {
                //获取数据
                TradeCourseBean next = input.iterator().next();

                String sst = DateFormatUtil.toYmdHms(window.getStart());
                String edt = DateFormatUtil.toYmdHms(window.getEnd());

                //添加字段
                next.setEdt(edt);
                next.setStt(sst);
                next.setTs(System.currentTimeMillis());

                //写出数据
                out.collect(next);
            }
        });


        //TODO 7.补充维表信息字段
        //关联course表
        SingleOutputStreamOperator<TradeCourseBean> courseDS =
                AsyncDataStream.unorderedWait(reduceDS, new DimAsyncFunction<TradeCourseBean>("DIM_COURSE_INFO") {
                    @Override
                    public String getKey(TradeCourseBean input) {
                        return input.getCourseId();
                    }

                    @Override
                    public void join(TradeCourseBean input, JSONObject dimInfo) {
                        input.setSubjectId(dimInfo.getString("SUBJECT_ID"));
                    }
                }, 100L, TimeUnit.SECONDS);


        //关联subject表
        SingleOutputStreamOperator<TradeCourseBean> subjectDS =
                AsyncDataStream.unorderedWait(courseDS, new DimAsyncFunction<TradeCourseBean>("DIM_SUBJECT") {
            @Override
            public String getKey(TradeCourseBean input) {
                return input.getSubjectId();
            }

            @Override
            public void join(TradeCourseBean input, JSONObject dimInfo) {
                input.setSubjectName(dimInfo.getString("SUBJECT_NAME"));
                input.setCategoryId(dimInfo.getString("CATEGORY_ID"));
            }
        },100L,TimeUnit.SECONDS);

        //关联category
        SingleOutputStreamOperator<TradeCourseBean> categoryDS = AsyncDataStream.unorderedWait(subjectDS,
                new DimAsyncFunction<TradeCourseBean>("DIM_CATEGORY_INFO") {
            @Override
            public String getKey(TradeCourseBean input) {
                return input.getCategoryId();
            }

            @Override
            public void join(TradeCourseBean input, JSONObject dimInfo) {
                input.setCategoryName(dimInfo.getString("CATEGORY_NAME"));
            }
        },100L,TimeUnit.SECONDS);

        //TODO 8.写入clickhouse
        categoryDS.print("=====>");
        //categoryDS.addSink(ClickHouseUtil.getJdbcSink("insert into dws_trade_course_order_window values(?,?,?,?,?,?,?,?,?,?,?,?,?,?)"));

        //TODO 9.执行任务
        env.execute("DwsTradeCourseOrderWindow");
    }
}
